import edu.rit.ds.Lease;
import edu.rit.ds.RemoteEventGenerator;
import edu.rit.ds.RemoteEventListener;
import edu.rit.ds.registry.AlreadyBoundException;
import edu.rit.ds.registry.NotBoundException;
import edu.rit.ds.registry.RegistryEvent;
import edu.rit.ds.registry.RegistryEventFilter;
import edu.rit.ds.registry.RegistryEventListener;
import edu.rit.ds.registry.RegistryProxy;
import java.io.IOException;
import java.rmi.NoSuchObjectException;
import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Queue;

/**
 * @author Alan Kaminsky
 * @source P2Pedia example
 * 
 * @contributor Praneeth Pulusani
 * Class MQServer provides a Java RMI distributed MQServer object in the Message Queuing system.
 * java Start MQServer <host> <port> <servername>
 * <host> is the name of the host computer where the Registry Server is running.
 * <port> is the port number to which the Registry Server is listening.
 * <servername> is the MQ server's name.
 */
public class MQServer implements MQServerRef {
	private String host;
	private int port;
	private String servername;
	int incoming = 0;
	int outgoing = 0;
	private HashMap<Integer, Queue<Message>> receiverQueue;
	private static RegistryEventListener registryListener;
	private static RegistryEventFilter registryFilter;
	private RemoteEventGenerator<MQServerEvent> eventGenerator;
	private RegistryProxy registry;

	/**
	 * Construct a new MQServer object.
	 * <P>
	 * The command line arguments are: <BR>
	 * <TT>args[0]</TT> = Registry Server's host <BR>
	 * <TT>args[1]</TT> = Registry Server's port <BR>
	 * <TT>args[2]</TT> =  MQ server's name.<BR>

	 * 
	 * @param args
	 *            Command line arguments.
	 * 
	 * @exception IllegalArgumentException
	 *                (unchecked exception) Thrown if there was a problem with
	 *                the command line arguments.
	 * @exception IOException
	 *                Thrown if an I/O error or a remote error occurred.
	 * @exception RemoteException
	 * 					Thrown when remote object cannot be reached
	 */
	public MQServer(String[] args) throws IOException, RemoteException {
		// Parse command line arguments.
		if (args.length != 3) {
			throw new IllegalArgumentException(
					"Incorrect number of arguments.\nUsage: java Start MQServer <host> <port> <servername>");
		}
		host = args[0];
		port = parseInt(args[1], "port");
		servername = args[2];

		receiverQueue = new HashMap<Integer, Queue<Message>>();
		// Prepare to generate remote events.
		eventGenerator = new RemoteEventGenerator<MQServerEvent>();

		// Get a proxy for the Registry Server.
		try {
			registry = new RegistryProxy(host, port);
		} catch (RemoteException e) {
			throw new RemoteException(
					"There is no Registry server running at given host and port");
		}
		// Export this MQServer.
		UnicastRemoteObject.exportObject(this, 0);

		// Bind this MQServer into the Registry Server.
		try {
			registry.bind(servername, this);
		} catch (AlreadyBoundException exc) {
			try {
				UnicastRemoteObject.unexportObject(this, true);
			} catch (NoSuchObjectException exc2) {
			}
			throw new IllegalArgumentException("\"" + servername
					+ "\" already exists");
		} catch (RemoteException exc) {
			try {
				UnicastRemoteObject.unexportObject(this, true);
			} catch (NoSuchObjectException exc2) {
			}
			throw exc;
		}
		registryListener = new RegistryEventListener() {
			public void report(long seqnum, RegistryEvent event) {
				processQueue(event.objectName());
			}

		};
		UnicastRemoteObject.exportObject(registryListener, 0);
		registryFilter = new RegistryEventFilter().reportType("Receiver")
				.reportBound();
		registry.addEventListener(registryListener, registryFilter);

	}

	/**
	 * Parse an integer command line argument.
	 * 
	 * @param arg
	 *            Command line argument.
	 * @param name
	 *            Argument name.
	 * 
	 * @return Integer value of <TT>arg</TT>.
	 * 
	 * @exception IllegalArgumentException
	 *                (unchecked exception) Thrown if <TT>arg</TT> cannot be
	 *                parsed as an integer.
	 */
	private static int parseInt(String arg, String name) {
		try {
			return Integer.parseInt(arg);
		} catch (NumberFormatException exc) {
			throw new IllegalArgumentException("MQServer(): Invalid <" + name
					+ ">: \"" + arg + "\"");
		}
	}

	/**
	 * Forward the message from sender to receiver.
	 * If the MQ server can't forward the message to the receiver 
	 * -- for example, if no receiver with the receiver ID exists, 
	 * or if there was a communication error while trying to forward the message 
	 * -- the MQ server adds the message to a first-in-first-out (FIFO) queue 
	 * associated with the receiver ID.
	 * If the MQ server can forward the message to the receiver, the MQ server does so, 
	 * and the message does not go into the queue.
	 * 
	 * @param senderID - ID of the sender
	 * @param reciverID - Place where message is going to
	 * @param Message - message that is being transported
	 * 
	 * @throws NotBoundException
	 * 
	 * @exception RemoteException
	 *                Thrown if a remote error occurred.
	 */
	public void sendMessage(final int senderID, final int receiverID,
			final String message) throws RemoteException, NotBoundException {
		try {
			incoming++;//keep track of number of incoming messages
			ReceiverRef receiver = (ReceiverRef) registry
					.lookup(this.servername + receiverID);
			receiver.loadMessage(senderID, message);
			outgoing++;//keep track of number of outgoing messages
			System.out.printf("From %s to %s: \"%s\" delivered\n", senderID,
					receiverID, message);
		} catch (NotBoundException e) {
			//If there is a receiver queue already, put it in the queue, otherwise 
			//create the queue and put in it.
			if (!receiverQueue.containsKey(receiverID)) {
				receiverQueue.put(receiverID, new LinkedList<Message>());
			}
			receiverQueue.get(receiverID).add(
					new Message(senderID, receiverID, message));
			System.out.printf("From %s to %s: \"%s\" queued\n", senderID,
					receiverID, message);
		}

		// Report a MQServerEvent to any remote event listeners.
		eventGenerator.reportEvent(new MQServerEvent(servername, incoming,
				outgoing));

	}

	/**
	 * Add the given remote event listener to this MQServer. Whenever a message is
	 * passed through to this server, this server will report a MQserverEvent to the given
	 * listener.
	 * 
	 * @param listener
	 *            Remote event listener.
	 */
	public Lease addListener(RemoteEventListener<MQServerEvent> listener)
			throws RemoteException {
		return eventGenerator.addListener(listener);
	}
	/**
	 * When a new receiver is added or an old receiver is respawned, MQServer gets a notification
	 * and processes the queued messages to deliver them
	 * @param receiverbindname
	 */
	private void processQueue(String receiverbindname) {
		try {
			ReceiverRef receiver = (ReceiverRef) registry
					.lookup(receiverbindname);

			Queue<Message> rQ = receiverQueue.get(receiver.getID());
			//check for null so that methods from a null object are not alled
			if (rQ != null) {
				//try to send messages from the queue one after the other
				while (!rQ.isEmpty()) {
					Message m = rQ.poll();
					receiver.loadMessage(m.senderID, m.message);
					outgoing++;
					eventGenerator.reportEvent(new MQServerEvent(servername, incoming,
							outgoing));
					System.out.printf("From %s to %s: \"%s\" delivered\n",
							m.senderID, m.receiverID, m.message);
				}
			}

		} catch (RemoteException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (NotBoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

	}
}
